package com.bcxj.Test;

import lombok.Data;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
 
// 假设OrderEvent是订单事件的类
public class OrderProcessing {
 
    // 定义一个ProcessFunction处理订单事件
    public static class OrderEventProcessor extends KeyedProcessFunction<String, OrderEvent, OrderResult> {
 
        private ValueState<Integer> orderCount;
 
        @Override
        public void open(Configuration parameters) throws Exception {
            orderCount = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("orderCount", Integer.class));
        }
 
        @Override
        public void processElement(OrderEvent orderEvent, Context context, Collector<OrderResult> collector) throws Exception {
            // 更新状态
            Integer currentCount = orderCount.value();
            if (currentCount == null) {
                currentCount = 0;
            }
            orderCount.update(currentCount + 1);
 
            // 输出处理结果
            OrderResult result = new OrderResult();
            result.setOrderId(orderEvent.getOrderId());
            result.setEventType(orderEvent.getEventType());
            result.setCount(currentCount);
            collector.collect(result);
        }
    }
 
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        DataStream<OrderEvent> orderEventStream = env.fromElements(
            // 示例订单事件数据
            new OrderEvent("1", "typeA"),
            new OrderEvent("2", "typeB"),
            new OrderEvent("1", "typeA")
            // ...
        );
 
        DataStream<OrderResult> resultStream = orderEventStream
            .keyBy(orderEvent -> orderEvent.getOrderId())
            .process(new OrderEventProcessor());
 
        resultStream.print();
 
        env.execute("Order Processing Job");
    }
}

@Data
// 订单事件类
class OrderEvent {
    private String orderId;
    private String eventType;

    public OrderEvent(String number, String typeA) {
        this.orderId = number;
        this.eventType = typeA;
    }
    // getters, setters, constructor
}

@Data
// 订单处理结果类
class OrderResult {
    private String orderId;
    private String eventType;
    private Integer count;
    // getters, setters
}